package demo04;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MovieClassifyReducer extends Reducer<Text, DistanceAndLabel, Text, NullWritable> {
	private int k=0;
	private static double a = 1.0; //a是曲线的高度
	private static double b = 0.0; //b是曲线中心线在x轴的偏移
	private static double c = 0.5; //半峰宽度（函数峰值一半处相距的宽度） 高斯方差
	
	@Override
	protected void setup(Reducer<Text, DistanceAndLabel, Text, NullWritable>.Context context)
			throws IOException, InterruptedException {
		//初始化K值
		k=context.getConfiguration().getInt("K",3);//找不到的话默认值为3
	}
	@Override
	protected void reduce(Text key, Iterable<DistanceAndLabel> value,
			Reducer<Text, DistanceAndLabel, Text, NullWritable>.Context context) throws IOException, InterruptedException {
		//100,0,35,17,95401,62,33,0,1,11,6,0,24,2,1,2,1,4,14,24,14,14,3     距离,类别
		String label=getWeight(getTopK(sort(value)));
		context.write(new Text(label+","+key), NullWritable.get());	
		//label(预测类型),100,0(实际类型),35,17,95401,62,33,0,1,11,6,0,24,2,1,2,1,4,14,24,14,14,3 
	}
	
	private Double weight(Double distance) {
		return a*Math.exp(-(Math.pow(distance-b,2))/(2*Math.pow(c, 2))) ;
	}
	/**
	 * 得到列表中类别的权重
	 * @param topK
	 * @return
	 */
	private String getWeight(List<String> topK) {
		//距离,类别
		HashMap<String,Double> labelTimes=new HashMap<String,Double>();// 类别，距离
		for (String str : topK) {
			Double distance=Double.parseDouble(str.substring(0,str.lastIndexOf(","))); //取距离
			String label=str.substring(str.lastIndexOf(",")+1,str.length()); //取类别
			if(labelTimes.containsKey(label)){
				labelTimes.put(label, labelTimes.get(label)+distance*weight(distance));//类别，距离*权重和
			}else{
				labelTimes.put(label, distance*weight(distance)); //初始化 labelTimes 类别，距离*权重
			}
		}
			
		Double maxDou=Double.MIN_VALUE; //判断
		String mostLabel=""; //决定最终是哪个类别
		for(Map.Entry<String, Double> kv:labelTimes.entrySet()){
			if(kv.getValue()>maxDou){
				maxDou=kv.getValue();
				mostLabel=kv.getKey();// key是类别  0是男 1是女
			}
		}
		return mostLabel;
	}
	
	/**
	 * 得到列表中类别的众数
	 * @param topK
	 * @return
	 */
	private String getMost(List<String> topK) {
		HashMap<String,Integer> labelTimes=new HashMap<String,Integer>();
		for (String str : topK) {
			String label=str.substring(str.lastIndexOf(",")+1,str.length()); //取类别
			if(labelTimes.containsKey(label)){
				labelTimes.put(label, labelTimes.get(label)+1);
			}else{
				labelTimes.put(label, 1);
			}
		}
		int maxInt=Integer.MIN_VALUE;
		String mostLabel="";
		for(Map.Entry<String, Integer> kv:labelTimes.entrySet()){
			if(kv.getValue()>maxInt){
				maxInt=kv.getValue();
				mostLabel=kv.getKey();// key是类别  0是男 1是女
			}
		}
		return mostLabel;
	}
	/**
	 * 取出列表中的前K个值
	 * @param sort   已被排序过的ArrayList
	 * @return 
	 */
	private List<String> getTopK(List<String> sort) {
		return sort.subList(0, k);
	}
	/**
	 * 根据距离升序排序
	 * @param value 距离,类别
	 * @return 
	 */
	private List<String> sort(Iterable<DistanceAndLabel> value) {
		ArrayList<String> result=new ArrayList<String>();
		for(DistanceAndLabel val:value){
			result.add(val.toString());
		}
		String[] tmp=new String[result.size()];//临时数据
		result.toArray(tmp);
		Arrays.sort(tmp, new Comparator<String>(){

			@Override
			public int compare(String o1, String o2) {
				double o1D=Double.parseDouble(o1.substring(0, o1.indexOf(",")));//取距离
				double o2D=Double.parseDouble(o2.substring(0, o2.indexOf(",")));
				if(o1D>o2D){
					return 1;
				}else if(o1D<o2D){
					return -1;
				}else{
					return 0;
				}
			}});
		return Arrays.asList(tmp);//回一个新的ArrayList
	}
}
